/*!
 * \file WtCtaEngine.cpp
 * \project	WonderTrader
 *
 * \author Wesley
 * \date 2020/03/30
 * 
 * \brief CTA交易引擎实现文件
 * 
 * 本文件实现了CTA(Commodity Trading Advisor)交易引擎的核心功能，
 * 包括策略管理、行情处理、信号生成、持仓管理、风控处理等。
 * CTA引擎是WonderTrader系统中用于商品期货交易的核心组件。
 */
#define WIN32_LEAN_AND_MEAN

#include "WtCtaEngine.h"
#include "WtDtMgr.h"
#include "WtCtaTicker.h"
#include "WtHelper.h"
#include "TraderAdapter.h"

#include "../Share/CodeHelper.hpp"
#include "../Includes/WTSVariant.hpp"
#include "../Share/TimeUtils.hpp"
#include "../Includes/IBaseDataMgr.h"
#include "../Includes/IHotMgr.h"
#include "../Includes/WTSContractInfo.hpp"
#include "../Includes/WTSRiskDef.hpp"
#include "../Share/decimal.h"

#include "../WTSTools/WTSLogger.h"

#include <rapidjson/document.h>
#include <rapidjson/prettywriter.h>
namespace rj = rapidjson;

#include <boost/asio.hpp>

boost::asio::io_context g_asyncIO;

/**
 * @brief CTA引擎构造函数
 * 
 * 初始化CTA引擎实例，设置行情处理器为空指针
 */
WtCtaEngine::WtCtaEngine()
	: _tm_ticker(NULL)
{
	
}

/**
 * @brief CTA引擎析构函数
 * 
 * 清理资源，释放行情处理器和配置对象
 */
WtCtaEngine::~WtCtaEngine()
{
	if (_tm_ticker)
	{
		delete _tm_ticker;
		_tm_ticker = NULL;
	}

	if (_cfg)
		_cfg->release();
}

/**
 * @brief 启动CTA引擎
 * 
 * 初始化行情处理器，生成标记文件，启动引擎运行
 * 
 * @param bAsync 是否异步运行，默认为false
 */
void WtCtaEngine::run(bool bAsync /* = false */)
{
	// 创建CTA实时行情处理器
	_tm_ticker = new WtCtaRtTicker(this);
	WTSVariant* cfgProd = _cfg->get("product");
	_tm_ticker->init(_data_mgr->reader(), cfgProd->getCString("session"));

	// 启动之前，先把所有的策略保存
	{
		rj::Document root(rj::kObjectType);
		rj::Document::AllocatorType &allocator = root.GetAllocator();

		// 保存策略列表
		rj::Value jStraList(rj::kArrayType);
		for (auto& m : _ctx_map)
		{
			const CtaContextPtr& ctx = m.second;
			jStraList.PushBack(rj::Value(ctx->name(), allocator), allocator);
		}

		root.AddMember("marks", jStraList, allocator);

		// 保存交易通道列表
		rj::Value jChnlList(rj::kArrayType);
		for (auto& m : _adapter_mgr->getAdapters())
		{
			const TraderAdapterPtr& adapter = m.second;
			jChnlList.PushBack(rj::Value(adapter->id(), allocator), allocator);
		}

		root.AddMember("channels", jChnlList, allocator);

		// 保存执行器列表
		rj::Value jExecList(rj::kArrayType);
		_exec_mgr.enum_executer([&jExecList, &allocator](ExecCmdPtr executer) {
			if(executer)
				jExecList.PushBack(rj::Value(executer->name(), allocator), allocator);
		});

		root.AddMember("executers", jExecList, allocator);

		root.AddMember("engine", rj::Value("CTA", allocator), allocator);

		// 写入标记文件
		std::string filename = WtHelper::getBaseDir();
		filename += "marker.json";

		rj::StringBuffer sb;
		rj::PrettyWriter<rj::StringBuffer> writer(sb);
		root.Accept(writer);
		StdFile::write_file_content(filename.c_str(), sb.GetString());
	}

	// 启动行情处理器
	_tm_ticker->run();

	// 启动风控监控器
	if (_risk_mon)
		_risk_mon->self()->run();

	// 如果不是异步模式，则阻塞运行
	if (!bAsync)
	{
		auto work = boost::asio::make_work_guard(g_asyncIO);
		g_asyncIO.run();
	}
}

/**
 * @brief 初始化CTA引擎
 * 
 * 设置基础数据管理器、数据管理器、热点管理器等核心组件
 * 
 * @param cfg 配置对象
 * @param bdMgr 基础数据管理器
 * @param dataMgr 数据管理器
 * @param hotMgr 热点管理器
 * @param notifier 事件通知器，默认为NULL
 */
void WtCtaEngine::init(WTSVariant* cfg, IBaseDataMgr* bdMgr, WtDtMgr* dataMgr, IHotMgr* hotMgr, EventNotifier* notifier /* = NULL */)
{
	WtEngine::init(cfg, bdMgr, dataMgr, hotMgr, notifier);

	_cfg = cfg;
	_cfg->retain();

	// 设置执行管理器的过滤管理器
	_exec_mgr.set_filter_mgr(&_filter_mgr);
}

/**
 * @brief 添加CTA策略上下文
 * 
 * 将策略上下文添加到引擎的策略映射表中
 * 
 * @param ctx CTA策略上下文智能指针
 */
void WtCtaEngine::addContext(CtaContextPtr ctx)
{
	uint32_t sid = ctx->id();
	_ctx_map[sid] = ctx;
}

/**
 * @brief 获取CTA策略上下文
 * 
 * 根据策略ID获取对应的策略上下文
 * 
 * @param id 策略ID
 * @return CtaContextPtr 策略上下文智能指针，如果不存在则返回空指针
 */
CtaContextPtr WtCtaEngine::getContext(uint32_t id)
{
	auto it = _ctx_map.find(id);
	if (it == _ctx_map.end())
		return CtaContextPtr();

	return it->second;
}

/**
 * @brief 引擎初始化回调
 * 
 * 在引擎初始化时调用，处理所有策略的初始化，
 * 计算目标持仓，应用风控规则，设置执行管理器的初始持仓
 */
void WtCtaEngine::on_init()
{
	faster_hashmap<LongKey, double> target_pos;
	
	// 遍历所有策略，调用初始化并收集目标持仓
	for (auto it = _ctx_map.begin(); it != _ctx_map.end(); it++)
	{
		CtaContextPtr& ctx = (CtaContextPtr&)it->second;
		ctx->on_init();

		// 枚举策略持仓，处理过滤规则和主力合约转换
		ctx->enum_position([this, &target_pos, ctx](const char* stdCode, double qty){

			double oldQty = qty;
			bool bFilterd = _filter_mgr.is_filtered_by_strategy(ctx->name(), qty);
			if (!bFilterd)
			{
				if (!decimal::eq(qty, oldQty))
				{
					// 输出日志
					WTSLogger::info_f("[Filters] Target position of {} of strategy {} reset by strategy filter: {} -> {}", 
						stdCode, ctx->name(), oldQty, qty);
				}

				// 处理主力合约和次主力合约代码转换
				std::string realCode = stdCode;
				if (CodeHelper::isStdFutHotCode(stdCode))
				{
					CodeHelper::CodeInfo cInfo = CodeHelper::extractStdCode(stdCode);
					std::string code = _hot_mgr->getRawCode(cInfo._exchg, cInfo._product, _cur_tdate);
					realCode = CodeHelper::rawMonthCodeToStdCode(code.c_str(), cInfo._exchg);
				}
				else if (CodeHelper::isStdFut2ndCode(stdCode))
				{
					CodeHelper::CodeInfo cInfo = CodeHelper::extractStdCode(stdCode);
					std::string code = _hot_mgr->getSecondRawCode(cInfo._exchg, cInfo._product, _cur_tdate);
					realCode = CodeHelper::rawMonthCodeToStdCode(code.c_str(), cInfo._exchg);
				}

				double& vol = target_pos[realCode];
				vol += qty;
			}
			else
			{
				// 输出日志
				WTSLogger::info("[Filters] Target position of %s of strategy %s ignored by strategy filter", stdCode, ctx->name());
			}
		});
	}

	// 检查是否启用风控
	bool bRiskEnabled = false;
	if (!decimal::eq(_risk_volscale, 1.0) && _risk_date == _cur_tdate)
	{
		WTSLogger::log_by_cat("risk", LL_INFO, "Risk scale of strategy group is %.2f", _risk_volscale);
		bRiskEnabled = true;
	}

	// 初始化持仓并打印日志
	for (auto it = target_pos.begin(); it != target_pos.end(); it++)
	{
		const auto& stdCode = it->first;
		double& pos = (double&)it->second;

		// 应用风控规模缩放
		if (bRiskEnabled && !decimal::eq(pos, 0))
		{
			double symbol = pos / abs(pos);
			pos = decimal::rnd(abs(pos)*_risk_volscale)*symbol;
		}

		WTSLogger::info_f("Portfolio initial position of {} is {}", stdCode.c_str(), pos);
	}

	// 设置执行管理器的初始持仓
	_exec_mgr.set_positions(target_pos);

	// 触发初始化事件
	if (_evt_listener)
		_evt_listener->on_initialize_event();
}

/**
 * @brief 交易时段开始回调
 * 
 * 在每个交易时段开始时调用，通知所有策略交易时段开始
 */
void WtCtaEngine::on_session_begin()
{
	WTSLogger::info("Trading day %u begun", _cur_tdate);
	
	// 通知所有策略交易时段开始
	for (auto it = _ctx_map.begin(); it != _ctx_map.end(); it++)
	{
		CtaContextPtr& ctx = (CtaContextPtr&)it->second;
		ctx->on_session_begin(_cur_tdate);
	}

	// 触发交易时段事件
	if (_evt_listener)
		_evt_listener->on_session_event(_cur_tdate, true);

	_ready = true;
}

/**
 * @brief 交易时段结束回调
 * 
 * 在每个交易时段结束时调用，通知所有策略交易时段结束
 */
void WtCtaEngine::on_session_end()
{
	WtEngine::on_session_end();

	// 通知所有策略交易时段结束
	for (auto it = _ctx_map.begin(); it != _ctx_map.end(); it++)
	{
		CtaContextPtr& ctx = (CtaContextPtr&)it->second;
		ctx->on_session_end(_cur_tdate);
	}

	WTSLogger::info("Trading day %u ended", _cur_tdate);
	
	// 触发交易时段事件
	if (_evt_listener)
		_evt_listener->on_session_event(_cur_tdate, false);
}

/**
 * @brief 定时调度回调
 * 
 * 定时调用所有策略的调度方法，重新计算目标持仓，
 * 应用过滤规则和风控规则，更新执行管理器
 * 
 * @param curDate 当前日期
 * @param curTime 当前时间
 */
void WtCtaEngine::on_schedule(uint32_t curDate, uint32_t curTime)
{
	// 重新加载过滤器
	_filter_mgr.load_filters();

	faster_hashmap<LongKey, double> target_pos;

	// 调用所有策略的调度方法
	for (auto it = _ctx_map.begin(); it != _ctx_map.end(); it++)
	{
		CtaContextPtr& ctx = (CtaContextPtr&)it->second;
		ctx->on_schedule(curDate, curTime);
		
		// 枚举策略持仓，处理过滤规则和主力合约转换
		ctx->enum_position([this, &target_pos, ctx](const char* stdCode, double qty){

			double oldQty = qty;
			bool bFilterd = _filter_mgr.is_filtered_by_strategy(ctx->name(), qty);
			if(!bFilterd)
			{
				if(!decimal::eq(qty, oldQty))
				{
					// 输出日志
					WTSLogger::info_f("[Filters] Target position of {} of strategy {} reset by strategy filter: {} -> {}", 
						stdCode, ctx->name(), oldQty, qty);
				}

				// 处理主力合约和次主力合约代码转换
				std::string realCode = stdCode;
				if (CodeHelper::isStdFutHotCode(stdCode))
				{
					CodeHelper::CodeInfo cInfo = CodeHelper::extractStdCode(stdCode);
					std::string code = _hot_mgr->getRawCode(cInfo._exchg, cInfo._product, _cur_tdate);
					realCode = CodeHelper::rawMonthCodeToStdCode(code.c_str(), cInfo._exchg);
				}
				else if (CodeHelper::isStdFut2ndCode(stdCode))
				{
					CodeHelper::CodeInfo cInfo = CodeHelper::extractStdCode(stdCode);
					std::string code = _hot_mgr->getSecondRawCode(cInfo._exchg, cInfo._product, _cur_tdate);
					realCode = CodeHelper::rawMonthCodeToStdCode(code.c_str(), cInfo._exchg);
				}

				double& vol = target_pos[realCode];
				vol += qty;
			}
			else
			{
				// 输出日志
				WTSLogger::info("[Filters] Target position of %s of strategy %s ignored by strategy filter", stdCode, ctx->name());
			}
		});
	}

	// 检查是否启用风控
	bool bRiskEnabled = false;
	if(!decimal::eq(_risk_volscale, 1.0) && _risk_date == _cur_tdate)
	{
		WTSLogger::log_by_cat("risk", LL_INFO, "Risk scale of strategy group is %.2f", _risk_volscale);
		bRiskEnabled = true;
	}

	// 处理目标持仓并应用风控
	for (auto it = target_pos.begin(); it != target_pos.end(); it++)
	{
		const auto& stdCode = it->first;
		double& pos = (double&)it->second;

		// 应用风控规模缩放
		if (bRiskEnabled && !decimal::eq(pos, 0))
		{
			double symbol = pos / abs(pos);
			pos = decimal::rnd(abs(pos)*_risk_volscale)*symbol;
		}

		// 添加信号到信号队列
		append_signal(stdCode.c_str(), pos, true);
	}

	// 检查是否有合约不在目标持仓中，需要平仓
	for(auto& m : _pos_map)
	{
		const auto& stdCode = m.first;
		if (target_pos.find(stdCode) == target_pos.end())
		{
			if(!decimal::eq(m.second._volume, 0))
			{
				// 如果不在目标持仓中，通知WtEngine去处理组合持仓清理
				append_signal(stdCode.c_str(), 0, true);

				WTSLogger::error("Instrument %s not in target positions, setup to 0 automatically", stdCode.c_str());
			}

			// 因为组合持仓中可能有过期的合约，这些合约不存在，所以在设置执行器之前要做一下检查
			auto cInfo = get_contract_info(stdCode.c_str());
			if(cInfo != NULL)
				target_pos[stdCode] = 0;
		}
	}

	// 异步更新资金动态盈亏
	push_task([this](){
		update_fund_dynprofit();
	});

	// 设置执行管理器的目标持仓
	_exec_mgr.set_positions(target_pos);

	// 保存数据
	save_datas();

	// 触发调度事件
	if (_evt_listener)
		_evt_listener->on_schedule_event(curDate, curTime);
}

/**
 * @brief 处理推送的行情数据
 * 
 * 将行情数据转发给行情处理器进行处理
 * 
 * @param newTick 新的行情数据
 * @param hotFlag 热点标志
 */
void WtCtaEngine::handle_push_quote(WTSTickData* newTick, uint32_t hotFlag)
{
	if (_tm_ticker)
		_tm_ticker->on_tick(newTick, hotFlag);
}

/**
 * @brief 处理持仓变化
 * 
 * 当策略持仓发生变化时调用，处理过滤规则、主力合约转换、
 * 风控规则，并通知执行管理器
 * 
 * @param straName 策略名称
 * @param stdCode 标准合约代码
 * @param diffQty 持仓变化量
 */
void WtCtaEngine::handle_pos_change(const char* straName, const char* stdCode, double diffQty)
{
	// 如果是持仓变化，可以不用处理未完成的单子，因为这是增量变化，不会改变目标diffQty
	if(_filter_mgr.is_filtered_by_strategy(straName, diffQty, true))
	{
		// 输出日志
		WTSLogger::info("[Filters] Target position of %s of strategy %s ignored by strategy filter", stdCode, straName);
		return;
	}

	// 处理主力合约和次主力合约代码转换
	std::string realCode = stdCode;
	if (CodeHelper::isStdFutHotCode(stdCode))
	{
		CodeHelper::CodeInfo cInfo = CodeHelper::extractStdCode(stdCode);
		std::string code = _hot_mgr->getRawCode(cInfo._exchg, cInfo._product, _cur_tdate);
		realCode = CodeHelper::rawMonthCodeToStdCode(code.c_str(), cInfo._exchg);
	}
	else if (CodeHelper::isStdFut2ndCode(stdCode))
	{
		CodeHelper::CodeInfo cInfo = CodeHelper::extractStdCode(stdCode);
		std::string code = _hot_mgr->getSecondRawCode(cInfo._exchg, cInfo._product, _cur_tdate);
		realCode = CodeHelper::rawMonthCodeToStdCode(code.c_str(), cInfo._exchg);
	}

	// 计算目标持仓
	PosInfo& pItem = _pos_map[realCode];
	double targetPos = pItem._volume + diffQty;

	// 检查是否启用风控
	bool bRiskEnabled = false;
	if (!decimal::eq(_risk_volscale, 1.0) && _risk_date == _cur_tdate)
	{
		WTSLogger::log_by_cat("risk", LL_INFO, "Risk scale of strategy group is %.2f", _risk_volscale);
		bRiskEnabled = true;
	}
	
	// 应用风控规模缩放
	if (bRiskEnabled && !decimal::eq(targetPos, 0))
	{
		double symbol = targetPos / abs(targetPos);
		targetPos = decimal::rnd(abs(targetPos)*_risk_volscale)*symbol;
	}

	// 添加信号并保存数据
	append_signal(realCode.c_str(), targetPos, false);
	save_datas();

	// 通知执行管理器处理持仓变化
	_exec_mgr.handle_pos_change(realCode.c_str(), targetPos);
}

/**
 * @brief 处理行情数据
 * 
 * 处理新的行情数据，转发给数据管理器和执行管理器，
 * 并根据订阅情况通知相关策略
 * 
 * @param stdCode 标准合约代码
 * @param curTick 当前行情数据
 */
void WtCtaEngine::on_tick(const char* stdCode, WTSTickData* curTick)
{
	WtEngine::on_tick(stdCode, curTick);

	// 转发行情给数据管理器
	_data_mgr->handle_push_quote(stdCode, curTick);

	// 转发行情给执行器，需要传递给执行器
	/*
	 *	这里不做任何判断，直接全部传递给执行器，主要是因为执行器可能会处理未完成的合约
	 *	需要注意的是，这里的合约可能过期
	 *	By Wesley @ 2021.08.19
	 */
	{
		// 不管是否是过期合约，都需要传递给执行器处理数据
		_exec_mgr.handle_tick(stdCode, curTick);
	}

	/*
	 *	By Wesley @ 2022.02.07
	 *	这里做了一个复权的调整
	 *	第一种情况订阅标记为0，即不复权模式，则直接按照原始代码触发ontick
	 *	第二种情况标记为1，即前复权模式，则将代码转成xxxx-，再次触发ontick
	 *	第三种情况标记为2，即后复权模式，则将代码转成xxxx+，再把tick数据做一个复权处理，再次触发ontick
	 */
	if(_ready)
	{
		auto sit = _tick_sub_map.find(stdCode);
		if (sit != _tick_sub_map.end())
		{
			const SubList& sids = sit->second;
			for (auto it = sids.begin(); it != sids.end(); it++)
			{
				uint32_t sid = it->first;
				

				auto cit = _ctx_map.find(sid);
				if (cit != _ctx_map.end())
				{
					CtaContextPtr& ctx = (CtaContextPtr&)cit->second;
					uint32_t opt = it->second.second;
					
					if (opt == 0)
					{
						// 不复权模式，直接触发
						ctx->on_tick(stdCode, curTick);
					}
					else
					{
						// 复权模式，需要处理代码和数据
						std::string wCode = stdCode;
						wCode = fmt::format("{}{}", stdCode, opt == 1 ? SUFFIX_QFQ : SUFFIX_HFQ);
						if (opt == 1)
						{
							// 前复权模式
							ctx->on_tick(wCode.c_str(), curTick);
						}
						else //(opt == 2)
						{
							// 后复权模式，需要调整价格数据
							WTSTickData* newTick = WTSTickData::create(curTick->getTickStruct());
							WTSTickStruct& newTS = newTick->getTickStruct();

							// 获取复权因子并调整价格
							double factor = _data_mgr->get_adjusting_factor(stdCode, get_trading_date());
							newTS.open *= factor;
							newTS.high *= factor;
							newTS.low *= factor;
							newTS.price *= factor;

							_price_map[wCode] = newTS.price;

							ctx->on_tick(wCode.c_str(), newTick);
							newTick->release();
						}
					}
					
				}

				
			}

		}
	}
	

	// 暂时先不计算成本，先简化处理
	/*
	double dynprofit = 0.0;
	for(auto v : _ctx_map)
	{
		const CtaContextPtr& ctx = v.second;
		dynprofit += ctx->get_dyn_profit();
	}

	WTSFundStruct& fundInfo = _port_fund->fundInfo();
	fundInfo._dynprofit = dynprofit;
	double dynbal = fundInfo._balance + dynprofit;
	if (fundInfo._max_dyn_bal != DBL_MAX)
		fundInfo._max_dyn_bal = std::max(fundInfo._max_dyn_bal, dynbal);
	else
		fundInfo._max_dyn_bal = dynbal;

	if (fundInfo._min_dyn_bal != DBL_MAX)
		fundInfo._min_dyn_bal = std::min(fundInfo._min_dyn_bal, dynbal);
	else
		fundInfo._min_dyn_bal = dynbal;
	*/
}

/**
 * @brief 处理K线数据
 * 
 * 当新的K线数据生成时调用，通知订阅了该K线的策略
 * 
 * @param stdCode 标准合约代码
 * @param period K线周期
 * @param times K线倍数
 * @param newBar 新的K线数据
 */
void WtCtaEngine::on_bar(const char* stdCode, const char* period, uint32_t times, WTSBarStruct* newBar)
{
	// 构建K线订阅键
	std::string key = StrUtil::printf("%s-%s-%u", stdCode, period, times);
	const SubList& sids = _bar_sub_map[key];
	
	// 通知所有订阅了该K线的策略
	for (auto it = sids.begin(); it != sids.end(); it++)
	{
		uint32_t sid = it->first;
		auto cit = _ctx_map.find(sid);
		if(cit != _ctx_map.end())
		{
			CtaContextPtr& ctx = (CtaContextPtr&)cit->second;
			ctx->on_bar(stdCode, period, times, newBar);
		}
	}

	WTSLogger::info("KBar [%s#%s%d] @ %u closed", stdCode, period, times, period[0] == 'd' ? newBar->date : newBar->time);
}

/**
 * @brief 检查是否在交易时间内
 * 
 * @return bool 如果在交易时间内返回true，否则返回false
 */
bool WtCtaEngine::isInTrading()
{
	return _tm_ticker->is_in_trading();
}

/**
 * @brief 将时间转换为分钟数
 * 
 * @param uTime 时间值
 * @return uint32_t 转换后的分钟数
 */
uint32_t WtCtaEngine::transTimeToMin(uint32_t uTime)
{
	return _tm_ticker->time_to_mins(uTime);
}

/**
 * @brief 获取品种信息
 * 
 * 根据标准合约代码获取对应的品种信息
 * 
 * @param stdCode 标准合约代码
 * @return WTSCommodityInfo* 品种信息指针
 */
WTSCommodityInfo* WtCtaEngine::get_comm_info(const char* stdCode)
{
	return _base_data_mgr->getCommodity(CodeHelper::stdCodeToStdCommID(stdCode).c_str());
}

/**
 * @brief 获取交易时段信息
 * 
 * 根据标准合约代码获取对应的交易时段信息
 * 
 * @param stdCode 标准合约代码
 * @return WTSSessionInfo* 交易时段信息指针，如果获取失败返回NULL
 */
WTSSessionInfo* WtCtaEngine::get_sess_info(const char* stdCode)
{
	WTSCommodityInfo* cInfo = _base_data_mgr->getCommodity(CodeHelper::stdCodeToStdCommID(stdCode).c_str());
	if (cInfo == NULL)
		return NULL;

	return _base_data_mgr->getSession(cInfo->getSession());
}

/**
 * @brief 获取实时时间
 * 
 * 获取当前的实时时间戳
 * 
 * @return uint64_t 实时时间戳
 */
uint64_t WtCtaEngine::get_real_time()
{
	return TimeUtils::makeTime(_cur_date, _cur_raw_time * 100000 + _cur_secs);
}